package MsgBus

import (
    "fmt"
    "time"

    "github.com/vmihailenco/msgpack"
)

type Subscriber struct {
    pipe   chan []byte
    id     int64
    topics []string
}

func newSubscriber(buffer_size int, id int64, topics []string) *Subscriber {
    x := &Subscriber{}
    x.pipe = make(chan []byte, buffer_size)
    x.id = id
    x.topics = append(x.topics, topics...)
    return x
}

func (s *Subscriber) ID() int64 {
    return s.id
}

func (s *Subscriber) Poll(timeout time.Duration) ([]byte, error) {
    if timeout == 0 {
        msg := <-s.pipe
        return msg, nil
    }
    tm := time.NewTimer(timeout)
    defer tm.Stop()
    select {
    case msg := <-s.pipe:
        return msg, nil

    case <-tm.C:
        return nil, fmt.Errorf("timeout")
    }
}

func (s *Subscriber) PollObject(obj interface{}, timeout time.Duration) error {
    b, err := s.Poll(timeout)
    if err != nil {
        return err
    }
    return msgpack.Unmarshal(b, obj)
}

func (s *Subscriber) Push(b []byte) error {
    select {
    case s.pipe <- b:
        return nil
    default:
        return fmt.Errorf("queue full")
    }
}

func (s *Subscriber) PushObject(obj interface{}) error {
    b, err := msgpack.Marshal(obj)
    if err != nil {
        return err
    }
    return s.Push(b)
}

func (s *Subscriber) markDeleted() {

}
